#pragma once
// 实现语音识别子服务
/*
    1.包含语音识别客户端
    2.包含brpc服务器，其中实现SpeechServiceImpl为调用语音客户端对语音信息进行识别
    3.服务器注册客户端etcd
*/
#include <brpc/server.h>
#include <butil/logging.h>
#include "logger.hpp"  //日志模块封装
#include "etcd.hpp"    //服务注册模块封装
#include "rabbitmq.hpp"
#include "transmite.pb.h"
#include "channel.hpp"
#include "base.pb.h"
#include "transmite.pb.h"
#include "user.pb.h"
#include "chat_session_member.hxx"
#include "chat_session_member-odb.hxx"
#include "utils.hpp"
#include "mysql_chat_session_member.hpp"

namespace zrb
{
    // （1）.创建rpc服务子类继承pb中的EchoService服务类，并实现内部的业务接口逻辑
    class TransmiteServiceImpl : public zrb::transmite_service
    {
    public:
        TransmiteServiceImpl(const std::string& user_service_name,
            const ServiceManager::ptr& mm_channel,
            const std::shared_ptr<odb::core::database>& mysql_odb,
            const MQClient::ptr& mq_client,
            const std::string& exchange,
            const std::string& routing_key)
            :_user_service_name(user_service_name)
            ,_mm_channel(mm_channel)
            ,_mysql_session_member_table(std::make_shared<ChatSessionMemberTable>(mysql_odb))
            ,_mq_client(mq_client) 
            ,_exchange(exchange)     
            ,_routing_key(routing_key) 
        {
        }

        ~TransmiteServiceImpl()
        {
        }

        void GetTransmitTarget(google::protobuf::RpcController *controller,
                  const ::zrb::NewMessageReq *request,
                  ::zrb::GetTransmitTargetRsp *response,
                  ::google::protobuf::Closure *done)
        {
            brpc::ClosureGuard rpc_guard(done);//在异步调用时，客户端不会阻塞等待响应，而是继续执行后续逻辑，当服务器返回响应后，brpc 会自动调用 `Done` 回调，让用户处理结果或错误。
            //定义一个错误回调函数
            auto err_response = [this,response](const std::string& id,const std::string& errmsg){
                response->set_request_id(id);
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };
            //1.进行消息的组织：发送者-用户子服务获取信息，所属会话、消息内容、产生时间、消息ID
            std::string uid = request->user_id();
            std::string rid = request->request_id();
            std::string ssid = request->chat_session_id();
            const MessageContent& content = request->message();
            //获取通信信道
            auto channel = _mm_channel->choose(_user_service_name); //获取信道
            if(!channel)
            {
                LOG_ERROR("未找到用户子服务节点");
                return err_response(rid,"未找到文件子服务模块");
            }
            //调用用户管理子服务，进行获取用户信息
            UserService_Stub stub(channel.get());
            GetUserInfoReq req; 

            //设置请求
            req.set_request_id(rid);
            req.set_user_id(uid);

            //发起请求
            brpc::Controller cntl;
            GetUserInfoRsp rsp;
            stub.GetUserInfo(&cntl, &req, &rsp, nullptr);

            if(cntl.Failed() || rsp.success() == false)
            {
                LOG_ERROR("{} 用户管理子服务调用失败 - {}",request->request_id(),cntl.ErrorText());
                return err_response(request->request_id(),"用户管理子服务调用失败!");
            }

            //进行消息的组织
            MessageInfo msg;
            msg.set_message_id(uuid());
            msg.set_chat_session_id(ssid); 
            msg.set_timestamp(time(nullptr));
            msg.mutable_sender()->CopyFrom(rsp.user_info());
            msg.mutable_message()->CopyFrom(content);

            //2.获取消息转发客户端用户列表
            auto user_lists = _mysql_session_member_table->member(ssid);
            //3.将封装完毕的消息，发布到消息队列，待消息存储子服务进行消息持久化
            bool ret = _mq_client->publish(_exchange,msg.SerializeAsString(),_routing_key);
            if(ret == false)
            {
                LOG_ERROR("{} 调用消息存储子服务失败进行消息持久化失败 - {}",request->request_id());
                return err_response(request->request_id(),"调用消息存储子服务失败进行消息持久化失败!");
            }
            std::cout << "正常运行555 " << ssid.size() << " " << ssid << std::endl;
            //4.组织响应
            response->set_request_id(rid);
            response->set_success(true);
            response->mutable_message()->CopyFrom(msg);
            for(const auto& user_id : user_lists)
            {
                response->add_target_id_list(user_id);
            }
            std::cout << "正常运行666" << std::endl;
        }
        private:
            std::string _user_service_name; //用户子服务的名称
            ServiceManager::ptr _mm_channel; //信道管理对象，用来与用户子服务进行通信

            ChatSessionMemberTable::ptr _mysql_session_member_table; //会话成员表的操作句柄

            MQClient::ptr _mq_client; //消息队列客户端

            std::string _exchange;     //交换机名称
            std::string _routing_key;  //binding_key
    };

    // （2）.创建rpc服务器类，搭建服务器
    //  （3）.向服务器中添加rpc子服务对象 -- 告诉服务器收到什么请求用哪个接口处理
    class TransmiteService
    {
    public:
        using ptr = shared_ptr<TransmiteService>;
        TransmiteService(std::shared_ptr<brpc::Server>& rpc_server,
            Registry::ptr& registry_client,
            Discovery::ptr& discovery_server,
            std::shared_ptr<odb::core::database>& mysql_odb)
            : _rpc_server(rpc_server)
            ,_registry_client(registry_client) 
            ,_discovery_server(discovery_server)
            ,_mysql_odb(mysql_odb)
        {
        }

        void start()
        {
            _rpc_server->RunUntilAskedToQuit();//休眠等待运行结束 
        }

        ~TransmiteService()
        {
        }

    private:
        std::shared_ptr<brpc::Server> _rpc_server;          // brpc服务器对象
        Registry::ptr _registry_client;                     // etcd服务器注册类
        Discovery::ptr _discovery_server;                   // etcd服务器发现类
        std::shared_ptr<odb::core::database> _mysql_odb;    // mysql数据库客户端
    };

    //SpeechServer构造过程参数太多，使用建造者模式
    class TransmiteServiceBuild
    {
    public:
        TransmiteServiceBuild()
        {}

        // 构造etcd服务注册对象
        // 网络Ip、注册的服务名称、注册的服务地址
        void make_reg_object(const std::string& host,const std::string& name,const std::string &access_host) 
        {
            _registry_client = std::make_shared<Registry>(host);
            _registry_client->registry(name,access_host); //服务注册
        }

        //构造mysql客户端
        void make_mysql_object(
            int max_pool,
            const std::string& user,
            const std::string& password,
            const std::string& dbname,
            const std::string& host,
            int port,
            const std::string& cset)
        {
            _mysql_odb = ODBFactory::create(max_pool,user,password,dbname,host,port,cset);
        }


        void make_mq_object(const std::string &user, 
            const std::string& passwd,
            const std::string& host,
            const std::string &exchange,
            const std::string &queue,
            const std::string& binding_key)
        {
            _exchange = exchange; 
            _routing_key = binding_key;
            _mq_client = std::make_shared<MQClient>(user,passwd,host);
            _mq_client->declareComponents(exchange,queue,binding_key);
        }

        // 构造brpc服务对象
        // 端口 超时时间 线程数量
        void make_rpc_object(uint16_t port, uint32_t timeout, uint8_t thread_num)
        {
            _rpc_server = std::make_shared<brpc::Server>();

            //向服务器增加一个SpeechServiceImpl服务，SpeechServiceImpl服务是自己实现的
            TransmiteServiceImpl* echo_service = new TransmiteServiceImpl(_user_service_name,
                _mm_channel,_mysql_odb,_mq_client,_exchange,_routing_key);
            //brpc::ServiceOwnership::SERVER_DOESNT_OWN_SERVICE 表示添加服务失败，服务器也不会删除服务对象
            //brpc::ServiceOwnership::SERVER_OWNS_SERVICE 表示添加服务失败时，服务器负责删除服务对象
            int ret = _rpc_server->AddService(echo_service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
            if (ret == -1) {
                LOG_ERROR("添加Rpc服务失败！");
                abort();
            }

            //启动服务器
            brpc::ServerOptions options;
            options.idle_timeout_sec = timeout; //连接空闲超时时间-超时后连接被关闭
            options.num_threads = thread_num; // io线程数量
            ret = _rpc_server->Start(port, &options); //监听的端口
            if (ret == -1) {
                LOG_ERROR("rpc服务器启动失败");
                abort();
            }
        }

        //用于构造服务发现客户端&信道管理对象
        void make_discovery_object(const std::string& reg_host,
            const std::string& base_service_name,
            const std::string& user_service_name)
        {
            _user_service_name = user_service_name;
            //信道管理对象
            _mm_channel = std::make_shared<ServiceManager>();
            _mm_channel->declared(user_service_name);
            auto put_cb = std::bind(&ServiceManager::onServiceOnline, _mm_channel.get(), std::placeholders::_1, std::placeholders::_2);
            auto del_cb = std::bind(&ServiceManager::onServiceOffline, _mm_channel.get(), std::placeholders::_1, std::placeholders::_2);
            //2. 构造服务发现对象
            _discovery_server = std::make_shared<Discovery>(reg_host, base_service_name, put_cb, del_cb);
        }
   
        TransmiteService::ptr build()
        {
            if (!_mysql_odb)
            {
                LOG_ERROR("还未初始化mysql数据库模块");
                abort();
            }   
            if(!_mm_channel)
            {
                LOG_ERROR("还未初始化信道管理服务模块");
                abort();
            }
            if(!_rpc_server)
            {
                LOG_ERROR("还未初始化rpc服务模块");
                abort();
            }
            if(!_discovery_server)
            {
                LOG_ERROR("还未初始化服务发现模块");
                abort();
            }
            if (!_registry_client)
            {
                LOG_ERROR("还未初始化服务注册客户端模块");
                abort();
            }
            if(!_mq_client)
            {
                LOG_ERROR("还未初始化服务注册消息队列客户端模块");
                abort();
            }

            TransmiteService::ptr server = std::make_shared<TransmiteService>(_rpc_server,
                _registry_client,_discovery_server,_mysql_odb);
            return server;
        }

    private:
        std::string _user_service_name;
        ServiceManager::ptr _mm_channel;
        std::shared_ptr<brpc::Server> _rpc_server;          // brpc服务器对象
        Registry::ptr _registry_client;                     // etcd服务器注册类
        Discovery::ptr _discovery_server;                   // etcd服务器发现类
        std::shared_ptr<odb::core::database> _mysql_odb;    // mysql数据库客户端
        MQClient::ptr _mq_client;                           // 消息队列客户端  
        std::string _exchange;                              //交换机名称
        std::string _routing_key;                           //bindingkey
    };
};
